package ml

import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{StringIndexer, StringIndexerModel, VectorAssembler}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object IRISTree_01 {
  def main(args: Array[String]): Unit = {
    //1 创建sparkSession
    val spark: SparkSession = SparkSession.builder().master("local[*]").appName("IRIS").getOrCreate()

    //导入隐式转化
    import spark.implicits._

    //2 读取文件数据
    val dataSource: DataFrame = spark.read.csv("file:///D:\\大数据\\学期文档\\项目\\03挖掘型标签\\数据集\\iris_tree.csv")
      .toDF("Sepal_Length", "Sepal_Width", "Petal_Length", "Petal_Width", "Species")
   // dataSource.show()
//      .select(
//        'Sepal_Length cast(DoubleType),
//        'Sepal_Width cast(DoubleType),
//        'Petal_Length cast(DoubleType),
//        'Petal_Width cast(DoubleType),
//        'Species
//      )
    dataSource.createOrReplaceTempView("dataSource")
    val dataSourceSQL: DataFrame = spark.sql(
      """
        |SELECT
        |CAST(Sepal_Length AS DOUBLE) AS Sepal_Length,
        |CAST(Sepal_Width AS DOUBLE) AS Sepal_Width,
        |CAST(Petal_Length AS DOUBLE) AS Petal_Length,
        |CAST(Petal_Width AS DOUBLE) AS Petal_Width,
        |Species
        |FROM dataSource
        |""".stripMargin)

    //3 标签处理 将字符的标签转为数字才能使用
    val Label: StringIndexerModel = new StringIndexer()
      .setInputCol("Species") //输入要转化位数字的字段
      .setOutputCol("label") //转化为数字后的字段名
      .fit(dataSourceSQL)
    //4 将特征数据转化为向量
    val features: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("Sepal_Length", "Sepal_Width", "Petal_Length", "Petal_Width"))
      .setOutputCol("features")

    //5 初始化决策树，进行分类
    val DecisionTree: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setPredictionCol("prediction")
      .setMaxDepth(5)

    //6 构建pipLine  传入stage
    val pipeline: Pipeline = new Pipeline()
      .setStages(Array(Label, features, DecisionTree))

    //7 将原始数据转化为训练数据和次测试数据
    val Array(dataTrain,dataTest): Array[Dataset[Row]] = dataSourceSQL.randomSplit(Array(0.8, 0.2))

    //8 使用pipline对训练数据进行训练，训练完后使用测试数据进行测试
    //使用训练数据训练得到转化器
    val model: PipelineModel = pipeline.fit(dataTrain)
    //使用测试数据进行测试
    val testDF: DataFrame = model.transform(dataTest)
    //testDF.show()

    //9 查看分类成功的百分比
    val evaluator: MulticlassClassificationEvaluator = new MulticlassClassificationEvaluator() //多类别评估器
      .setLabelCol("label") //设置原始数据的label
      .setPredictionCol("prediction")//设置根据数据算出来的结果
    val score: Double = evaluator.evaluate(testDF)
   // println(score)

    //10 决策树过程
    val decisionTreeClassificationModel: DecisionTreeClassificationModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]
    println(decisionTreeClassificationModel.toDebugString)

  }
}
